package org.dreamwork.tools.concurrent;

import org.dreamwork.tools.LocalUtil;
import org.dreamwork.tools.LogFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.logging.Logger;

/**
 * Created by seth.yang on 2018/12/10
 */
public class SimpleBroadcaster {
    private static final Object LOCK = new byte[0];
    private static final Object QUIT = new byte[0];
    private static final BlockingQueue<Object> queue = new LinkedBlockingQueue<> ();
    private static final Map<String, List<IReceiver>> receivers = new ConcurrentHashMap<> ();
    private static final SimpleBroadcaster instance = new SimpleBroadcaster ();

    private static final Logger logger = LogFactory.getLogger (SimpleBroadcaster.class);

    public static SimpleBroadcaster getInstance () {
        return instance;
    }

    private static final class Wrapper {
        String category;
        Message message;

        Wrapper (String category, Message message) {
            this.category = category;
            this.message = message;
        }
    }

    static {
        LocalUtil.service.execute (()->{
            logger.info ("simple broadcaster start.");
            while (true) {
                Object o;
                try {
                    o = queue.take ();

                    if (o == QUIT) {
                        break;
                    }

                    Wrapper w = (Wrapper) o;
                    Message message  = w.message;
                    String  category = w.category;
                    final List<IReceiver> copy;
                    synchronized (LOCK) {
                        if (receivers.containsKey (category)) {
                            copy = new ArrayList<> (receivers.get (category));
                        } else {
                            copy = null;
                        }
                    }

                    if (copy != null && !copy.isEmpty ()) {
                        for(IReceiver receiver : copy) {
                            receiver.received (message);
                        }
                    }
                } catch (InterruptedException ex) {
                    ex.printStackTrace ();
                }
            }
        });
    }

    public void register (String category, IReceiver receiver) {
        synchronized (LOCK) {
            List<IReceiver> list = receivers.computeIfAbsent (category, key->new ArrayList<> ());
            if (!list.contains (receiver)) {
                list.add (receiver);
            }
        }
    }

    public void unregister (String category, IReceiver receiver) {
        synchronized (LOCK) {
            if (receivers.containsKey (category)) {
                List<IReceiver> list = receivers.get (category);
                list.remove (receiver);

                if (list.isEmpty ()) {
                    receivers.remove (category);
                }
            }
        }
    }

    public void broadcast (String category, int what) {
        broadcast (category, new Message (what));
    }

    public void broadcast (String category, Object arg) {
        broadcast (category, new Message (arg));
    }

    public void broadcast (String category, int what, Object arg) {
        broadcast (category, new Message (what, arg));
    }

    public void broadcast (String category, Message message) {
        queue.offer (new Wrapper (category, message));
    }

    public void shutdown () {
        queue.offer (QUIT);
    }
}